
package com.lee.service; 
 
import com.alibaba.fastjson.JSON; 
import com.alibaba.fastjson.JSONObject; 
import com.lee.fpflink.CountIndexFunction; 
import com.lee.fpflink.DataClassifyKeyByFunction; 
import com.lee.fpflink.FindFrequentWindowFunction; 
import com.lee.fpflink.GenerateRulesWindowFunction; 
import com.lee.info.CkFlowInfo; 
import com.lee.info.UserFlowInfo; 
import com.lee.logger.Logger; 
import com.lee.source.HbaseSource; 
import com.lee.tree.FPStreamGenerate; 
import com.lee.tree.PatternTree; 
import com.lee.tree.TwoTuple; 
import com.lee.utils.*; 
import org.apache.flink.api.common.functions.FlatMapFunction; 
import org.apache.flink.api.common.serialization.SimpleStringSchema; 
import org.apache.flink.api.common.typeinfo.TypeHint; 
import org.apache.flink.api.common.typeinfo.TypeInformation; 
import org.apache.flink.streaming.api.datastream.DataStream; 
import org.apache.flink.streaming.api.datastream.DataStreamSource; 
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 
import org.apache.flink.streaming.api.functions.windowing.WindowFunction; 
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow; 
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; 
import org.apache.flink.util.Collector; 
 
import java.util.*; 
 
public class FPFlinkStream { 
    public static void main(String[] args) throws Exception {
 
        Logger.info("FPFlink start0504 " + ConstantsUtils.MIN_SUPPORT);
 
        //从kafka获取数据
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers","localhost:9092");
        properties.setProperty("group.id","consumer-group");
        properties.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("auto.offset.reset","latest");
 
 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        FlinkKafkaConsumer<String> marketingConsumer = new FlinkKafkaConsumer<>("marketing", new SimpleStringSchema(), properties);
 
        DataStreamSource<String> source = env.addSource(marketingConsumer);
        //打印获取的数据
        source.print();
        Logger.info("get tag");
        //将json数据转换为对象
        DataStream<UserFlowInfo> dataStream = source.flatMap((FlatMapFunction<String, UserFlowInfo>) (line, collector) -> {
            if("command:init".equals(line)){
                DMapUtils.initMap();
                DCountUtils.initMap();
                SupportDataUtils.initMap();
                RulesUtils.initMap();
                CountsUtils.initMap();
                CrowdFlowUtils.initMap();
                return ;
            }
            JSONObject jsonObject;
            try {
                jsonObject = JSON.parseObject(line);
                UserFlowInfo userFlowInfo = new UserFlowInfo();
                userFlowInfo.setCustId(String.valueOf(jsonObject.get("cust_id")));
                userFlowInfo.setName(String.valueOf(jsonObject.get("name")));
                userFlowInfo.setActivityCode(String.valueOf(jsonObject.get("activity_code")));
                userFlowInfo.setActivityName(String.valueOf(jsonObject.get("activity_name")));
                userFlowInfo.setTime(String.valueOf(jsonObject.get("time")));
 
                collector.collect(userFlowInfo);
            }catch (Exception e){
                Logger.error("错误的json格式数据",line,e);
                return ;
 
            }
 
        }).returns(TypeInformation.of(new TypeHint<UserFlowInfo>() {}));
 
 
        //获取用户的tag
        DataStream<List<String>> rawStream = dataStream.flatMap((FlatMapFunction<UserFlowInfo, List<String>>) (item, collector) ->{
            if(CrowdFlowUtils.judgeKeyInMap(item.getCustId(),item.getTime())){
                Logger.error("当前cust_id存在在hbase中，跳过，custId ",item.getCustId());
                return ;
            }
            CrowdFlowUtils.setValueByKey(item.getCustId(),item.getName(),item.getActivityCode(),item.getActivityName(),item.getTime());
 
            String custId = item.getCustId();
            String activityCode = item.getActivityCode();
            List<String> crowdInfo = HbaseSource.getCrowdInfoMap(custId,activityCode);
//            Logger.info(crowdInfo);
            if(crowdInfo != null && crowdInfo.size() > 0){
                collector.collect(crowdInfo);
            } else {
                return ;
            }
 
        }).returns(TypeInformation.of(new TypeHint<List<String>>() {}));
        Logger.info("convert");
 
 
 
        rawStream.flatMap(new CountIndexFunction())
                .keyBy(new DataClassifyKeyByFunction())
                .countWindow(ConstantsUtils.DATA_BLOCK_SIZE).apply(new FindFrequentWindowFunction()).setParallelism(8)
                .flatMap(new GenerateRulesWindowFunction()).setParallelism(1);
 
 
 
        env.execute();
    }
 
 
 
 
 
} 
